… to_protobuf

From SandishKumarHN(sanysandishgmail.com) and Mohan Parthasarathy(mposdev21gmail.com)

This PR follows main PR #37972

The following is an example of how to use from_protobuf and to_protobuf in Pyspark.

```python
data = [("1", (2, "Alice", 109200))]
ddl_schema = "key STRING, value STRUCT<age: INTEGER, name: STRING, score: LONG>"
df = spark.createDataFrame(data, ddl_schema)
desc_hex = str('0ACE010A41636F6E6E6563746F722F70726F746F6275662F7372632F746573742F726'
... '5736F75726365732F70726F746F6275662F7079737061726B5F746573742E70726F746F121D6F72672E61'
... '70616368652E737061726B2E73716C2E70726F746F627566224B0A0D53696D706C654D657373616765121'
... '00A03616765180120012805520361676512120A046E616D6518022001280952046E616D6512140A057363'
... '6F7265180320012803520573636F72654215421353696D706C654D65737361676550726F746F736206707'
... '26F746F33')
 import tempfile
# Writing a protobuf description into a file, generated by using
# connector/protobuf/src/test/resources/protobuf/pyspark_test.proto file
with tempfile.TemporaryDirectory() as tmp_dir:
...     desc_file_path = "%s/pyspark_test.desc" % tmp_dir
...     with open(desc_file_path, "wb") as f:
...         _ = f.write(bytearray.fromhex(desc_hex))
...         f.flush()
...         message_name = 'SimpleMessage'
...         proto_df = df.select(to_protobuf(df.value,
...         desc_file_path, message_name).alias("value"))
...         proto_df.show(truncate=False)
...         proto_df = proto_df.select(from_protobuf(proto_df.value,
...         desc_file_path, message_name).alias("value"))
...         proto_df.show(truncate=False)
    +----------------------------------------+
    |value                                   |
    +----------------------------------------+
    |[08 02 12 05 41 6C 69 63 65 18 90 D5 06]|
    +----------------------------------------+
    +------------------+
    |value             |
    +------------------+
    |{2, Alice, 109200}|
    +------------------+
```

### ****Tests Covered****
- from_protobuf / to_protobuf (functions.py)

Closes #38212 from SandishKumarHN/PYSPARK_PROTOBUF.

Authored-by: SandishKumarHN <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

…protobuf

From SandishKumarHN(sanysandishgmail.com) and Mohan Parthasarathy(mposdev21gmail.com)

Protocol buffers are Google's language-neutral, platform-neutral, extensible mechanism for serializing structured data. It is widely used in Kafka-based data pipelines. Unlike Avro, Spark does not have native support for protobuf. This PR provides two new functions from_protobuf/to_protobuf to read and write Protobuf data within a data frame.

The implementation is closely modeled after Avro implementation so that it is easy to understand and review the changes.

Following is an example of typical usage.

```scala
// `from_proto` requires absolute path of Protobuf schema file
// and the protobuf message within the file
val userProtoFile = "./examples/src/main/resources/user.desc"
val userProtoMsg = "User"

val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "proto-topic-in")
  .load()

// 1. Decode the Protobuf data into a struct;
// 2. Filter by column `favorite_color`;
// 3. Encode the column `name` in Protobuf format.
val output = df
  .select(from_protobuf('value, userProtoFile, userProtoMsg) as 'user)
  .where("user.favorite_color == \"red\"")
  .select(to_protobuf($"user.name", userProtoFile, userProtoMsg) as 'value)

val query = output
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "proto-topic-out")
  .start()
```

The new functions are very similar to Avro

- from_protobuf requires the proto descriptor file and the message type within that file which is similar to from_avro requiring the JSON schema.
- to_protobuf is similar to to_avro and does not require the proto descriptor file as it can build the schema (protobuf descriptor) from the catalyst types. Similarly, to_proto (like to_avro) can also take in the descriptor file for describing the schema

- Protobuf format proto3 is supported ( Even though proto2 and proto3 are inter-operable, we have explicitly tested only with proto3)
- Google protobuf supported types
    - Scalar value types
    - Enumerations
    - Message types as field types
    - Nested Messages
    - Maps
    - Unknown fields are well-formed protocol buffer serialized data representing fields that the parser does not recognize. Original version of proto3 did not include this when there are parsing problems. This feature is needed to detect schemas that does not match the message type and needed to support FAIL_SAFE and PERMISSIVE mode. This feature is available in proto3 with version. 3.5 onwards

- Any requires the knowledge of the underlying object type when deserializing the message and generally not considered type safe
- OneOf requires the knowledge of the object type that was encoded when deserializing the message
- Custom Options is an advanced feature within protobuf where the users can define their own options
- Catalyst types that are not natively supported in protobuf. This happens normally during serialization and an exception will be thrown when following types are encountered
    - DecimalType
    - DateType
    - TimestampType

Tests have been written to test at different levels

- from_protobuf / to_protobuf (ProtoFunctionSuite)
- ProtoToCatalyst / CatalystToProto (ProtoCatalystDataConversionSuite)
- ProtoDeserializer / ProtoSerializer (ProtoSerdeSuite)

A bunch of roundtrip tests go through to_protobuf(from_proto) or from_protobuf(to_proto) and compare the results. It also repeats some of the tests where to_protobuf is called without a descriptor file where the protobuf descriptor is built from the catalyst types.

- roundtrip in to_protobuf and from_protobuf for struct for protobuf scalar types
- roundtrip in to_protobuf(without descriptor params) and from_proto - struct for protobuf scalar types
- roundtrip in from_protobuf and to_protobuf - Repeated protobuf types
- roundtrip in from_protobuf and to_protobuf - Repeated Message Once
- roundtrip in from_protobuf and to_protobuf - Repeated Message Twice
- roundtrip in from_protobuf and to_protobuf - Map
- roundtrip in from_protobuf and to_protobuf - Enum
- roundtrip in from_protobuf and to_protobuf - Multiple Message
- roundtrip in to_protobuf and from_protobuf - with null

- Test basic conversion - serialize(deserialize(message)) == message
- Fail to convert with field type mismatch - Make sure the right exception is thrown for incompatible schema for serializer and deserializer
- Fail to convert with missing nested Protobuf fields
- Fail to convert with deeply nested field type mismatch
- Fail to convert with missing Catalyst fields

- ProtoToCatalyst(to_protobuf(basic_catalyst_types )): Boolean,Integer,Double,Float,Binary,String,Byte,Shost
- Handle unsupported input of Message type: Serialize a message first and deserialize using a bad schema. Test with FAILFAST to get an exception and PERMISSIVE to get a null row
- filter push-down to proto deserializer: Filtering the rows based on the filter during proto deserialization
- Test ProtoDeserializer with binary message type

Recent(10-04-2022) changes have been tested with the configurations listed below.
Job: Kafka + Spark Structure Streaming
2 executors, each with 2048m and 2 cores
150-200 events/second each event having 100 fields(basic types, message, map type, enum)

Closes apache#37972 from SandishKumarHN/SPARK_PROTO_1.

Lead-authored-by: SandishKumarHN <[email protected]>
Co-authored-by: Sandish Kumar Hebbani Naga <[email protected]>
Co-authored-by: Mohan Parthasarathy <[email protected]>
Co-authored-by: sandishkumarhn <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>

… to_protobuf

From SandishKumarHN(sanysandishgmail.com) and Mohan Parthasarathy(mposdev21gmail.com)

This PR follows main PR apache#37972

The following is an example of how to use from_protobuf and to_protobuf in Pyspark.

```python
data = [("1", (2, "Alice", 109200))]
ddl_schema = "key STRING, value STRUCT<age: INTEGER, name: STRING, score: LONG>"
df = spark.createDataFrame(data, ddl_schema)
desc_hex = str('0ACE010A41636F6E6E6563746F722F70726F746F6275662F7372632F746573742F726'
... '5736F75726365732F70726F746F6275662F7079737061726B5F746573742E70726F746F121D6F72672E61'
... '70616368652E737061726B2E73716C2E70726F746F627566224B0A0D53696D706C654D657373616765121'
... '00A03616765180120012805520361676512120A046E616D6518022001280952046E616D6512140A057363'
... '6F7265180320012803520573636F72654215421353696D706C654D65737361676550726F746F736206707'
... '26F746F33')
 import tempfile
with tempfile.TemporaryDirectory() as tmp_dir:
...     desc_file_path = "%s/pyspark_test.desc" % tmp_dir
...     with open(desc_file_path, "wb") as f:
...         _ = f.write(bytearray.fromhex(desc_hex))
...         f.flush()
...         message_name = 'SimpleMessage'
...         proto_df = df.select(to_protobuf(df.value,
...         desc_file_path, message_name).alias("value"))
...         proto_df.show(truncate=False)
...         proto_df = proto_df.select(from_protobuf(proto_df.value,
...         desc_file_path, message_name).alias("value"))
...         proto_df.show(truncate=False)
    +----------------------------------------+
    |value                                   |
    +----------------------------------------+
    |[08 02 12 05 41 6C 69 63 65 18 90 D5 06]|
    +----------------------------------------+
    +------------------+
    |value             |
    +------------------+
    |{2, Alice, 109200}|
    +------------------+
```

- from_protobuf / to_protobuf (functions.py)

Closes apache#38212 from SandishKumarHN/PYSPARK_PROTOBUF.

Authored-by: SandishKumarHN <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

…lasses

This is the follow-up PR to #37972 and #38212

### What changes were proposed in this pull request?
1. Move spark-protobuf error classes to the spark error-classes framework(core/src/main/resources/error/error-classes.json).
2. Support protobuf imports
3. validate protobuf timestamp and duration types.

### Why are the changes needed?
N/A

### Does this PR introduce _any_ user-facing change?
None

### How was this patch tested?
Existing tests should cover the validation of this PR.

CC: rangadi mposdev21 gengliangwang

Closes #38344 from SandishKumarHN/SPARK-40777-ProtoErrorCls.

Authored-by: SandishKumarHN <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>

… to_protobuf

From SandishKumarHN(sanysandishgmail.com) and Mohan Parthasarathy(mposdev21gmail.com)

This PR follows main PR apache#37972

The following is an example of how to use from_protobuf and to_protobuf in Pyspark.

```python
data = [("1", (2, "Alice", 109200))]
ddl_schema = "key STRING, value STRUCT<age: INTEGER, name: STRING, score: LONG>"
df = spark.createDataFrame(data, ddl_schema)
desc_hex = str('0ACE010A41636F6E6E6563746F722F70726F746F6275662F7372632F746573742F726'
... '5736F75726365732F70726F746F6275662F7079737061726B5F746573742E70726F746F121D6F72672E61'
... '70616368652E737061726B2E73716C2E70726F746F627566224B0A0D53696D706C654D657373616765121'
... '00A03616765180120012805520361676512120A046E616D6518022001280952046E616D6512140A057363'
... '6F7265180320012803520573636F72654215421353696D706C654D65737361676550726F746F736206707'
... '26F746F33')
 import tempfile
# Writing a protobuf description into a file, generated by using
# connector/protobuf/src/test/resources/protobuf/pyspark_test.proto file
with tempfile.TemporaryDirectory() as tmp_dir:
...     desc_file_path = "%s/pyspark_test.desc" % tmp_dir
...     with open(desc_file_path, "wb") as f:
...         _ = f.write(bytearray.fromhex(desc_hex))
...         f.flush()
...         message_name = 'SimpleMessage'
...         proto_df = df.select(to_protobuf(df.value,
...         desc_file_path, message_name).alias("value"))
...         proto_df.show(truncate=False)
...         proto_df = proto_df.select(from_protobuf(proto_df.value,
...         desc_file_path, message_name).alias("value"))
...         proto_df.show(truncate=False)
    +----------------------------------------+
    |value                                   |
    +----------------------------------------+
    |[08 02 12 05 41 6C 69 63 65 18 90 D5 06]|
    +----------------------------------------+
    +------------------+
    |value             |
    +------------------+
    |{2, Alice, 109200}|
    +------------------+
```

### ****Tests Covered****
- from_protobuf / to_protobuf (functions.py)

Closes apache#38212 from SandishKumarHN/PYSPARK_PROTOBUF.

Authored-by: SandishKumarHN <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

…lasses

This is the follow-up PR to apache#37972 and apache#38212

### What changes were proposed in this pull request?
1. Move spark-protobuf error classes to the spark error-classes framework(core/src/main/resources/error/error-classes.json).
2. Support protobuf imports
3. validate protobuf timestamp and duration types.

### Why are the changes needed?
N/A

### Does this PR introduce _any_ user-facing change?
None

### How was this patch tested?
Existing tests should cover the validation of this PR.

CC: rangadi mposdev21 gengliangwang

Closes apache#38344 from SandishKumarHN/SPARK-40777-ProtoErrorCls.

Authored-by: SandishKumarHN <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>

…nition from `protobuf`

### What changes were proposed in this pull request?
This PR removes the unused object definition `ScalaReflectionLock` from the `protobuf` module. `ScalaReflectionLock` is a definition at the access scope of `protobuf` package, which was defined in SPARK-40654 | #37972 and become unused in SPARK-41639 | #39147.

### Why are the changes needed?
Clean up unused definitions.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Pass GitHub Actions

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #47459 from LuciferYang/remove-ScalaReflectionLock.

Authored-by: yangjie01 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

…nition from `protobuf`

### What changes were proposed in this pull request?
This PR removes the unused object definition `ScalaReflectionLock` from the `protobuf` module. `ScalaReflectionLock` is a definition at the access scope of `protobuf` package, which was defined in SPARK-40654 | apache#37972 and become unused in SPARK-41639 | apache#39147.

### Why are the changes needed?
Clean up unused definitions.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Pass GitHub Actions

### Was this patch authored or co-authored using generative AI tooling?
No

Closes apache#47459 from LuciferYang/remove-ScalaReflectionLock.

Authored-by: yangjie01 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

…nition from `protobuf`

### What changes were proposed in this pull request?
This PR removes the unused object definition `ScalaReflectionLock` from the `protobuf` module. `ScalaReflectionLock` is a definition at the access scope of `protobuf` package, which was defined in SPARK-40654 | apache#37972 and become unused in SPARK-41639 | apache#39147.

### Why are the changes needed?
Clean up unused definitions.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Pass GitHub Actions

### Was this patch authored or co-authored using generative AI tooling?
No

Closes apache#47459 from LuciferYang/remove-ScalaReflectionLock.

Authored-by: yangjie01 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

…nition from `protobuf`

### What changes were proposed in this pull request?
This PR removes the unused object definition `ScalaReflectionLock` from the `protobuf` module. `ScalaReflectionLock` is a definition at the access scope of `protobuf` package, which was defined in SPARK-40654 | apache#37972 and become unused in SPARK-41639 | apache#39147.

### Why are the changes needed?
Clean up unused definitions.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Pass GitHub Actions

### Was this patch authored or co-authored using generative AI tooling?
No

Closes apache#47459 from LuciferYang/remove-ScalaReflectionLock.

Authored-by: yangjie01 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

…nition from `protobuf`

### What changes were proposed in this pull request?
This PR removes the unused object definition `ScalaReflectionLock` from the `protobuf` module. `ScalaReflectionLock` is a definition at the access scope of `protobuf` package, which was defined in SPARK-40654 | apache#37972 and become unused in SPARK-41639 | apache#39147.

### Why are the changes needed?
Clean up unused definitions.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Pass GitHub Actions

### Was this patch authored or co-authored using generative AI tooling?
No

Closes apache#47459 from LuciferYang/remove-ScalaReflectionLock.

Authored-by: yangjie01 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>